package com.fsxgt.datax;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.datax.common.constant.PluginType;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.AbstractJobPlugin;
import com.alibaba.datax.common.plugin.AbstractPlugin;
import com.alibaba.datax.common.plugin.AbstractTaskPlugin;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.taskgroup.runner.AbstractRunner;
import com.alibaba.datax.core.taskgroup.runner.ReaderRunner;
import com.alibaba.datax.core.taskgroup.runner.WriterRunner;
import com.alibaba.datax.core.util.FrameworkErrorCode;
import com.alibaba.datax.core.util.container.JarLoader;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;

/**
 * Created by jingxing on 14-8-24.
 * <p/>
 * 插件加载器，大体上分reader、transformer（还未实现）和writer三中插件类型，
 * reader和writer在执行时又可能出现Job和Task两种运行时（加载的类不同）
 */
public class LoadUtil {
	
	private static final Logger LOG = LoggerFactory.getLogger(LoadUtil.class);



    private LoadUtil() {
    }

    private enum ContainerType {
        Job("Job"), Task("Task");
        private String type;

        private ContainerType(String type) {
            this.type = type;
        }

        public String value() {
            return type;
        }
    }




    public static Configuration getPluginConf(String pluginType, String pluginName) {

    	//
    	//Configuration pluginConf = pluginRegisterCenter
    	//.getConfiguration(generatePluginKey(pluginType, pluginName));

    			String path = String.format("com/alibaba/datax/plugin/%s/%s/plugin.json", pluginType, pluginName);

    			Configuration pluginConf = Configuration.from(LoadUtil.class.getClassLoader().getResourceAsStream(path));

    			if (null == pluginConf) {
    				
    			    LOG.info("getPluginConf:"+path);
    				
    				throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INSTALL_ERROR,
    						String.format("DataX不能找到插件[%s]的配置.", pluginName));
    			}

    			return pluginConf;
    }

   

    /**
     * 加载taskPlugin，reader、writer都可能加载
     *
     * @param pluginType
     * @param pluginName
     * @return
     */
    public static AbstractTaskPlugin loadTaskPlugin(PluginType pluginType,
                                                    String pluginName) {
        Class<? extends AbstractPlugin> clazz = LoadUtil.loadPluginClass(
                pluginType, pluginName, ContainerType.Task);

        try {
            AbstractTaskPlugin taskPlugin = (AbstractTaskPlugin) clazz
                    .newInstance();
            taskPlugin.setPluginConf(getPluginConf(pluginType.toString(), pluginName));
            return taskPlugin;
        } catch (Exception e) {
            throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,
                    String.format("DataX不能找plugin[%s]的Task配置.",
                            pluginName), e);
        }
    }

    /**
     * 根据插件类型、名字和执行时taskGroupId加载对应运行器
     *
     * @param pluginType
     * @param pluginName
     * @return
     */
    public static AbstractRunner loadPluginRunner(PluginType pluginType, String pluginName) {
        AbstractTaskPlugin taskPlugin = LoadUtil.loadTaskPlugin(pluginType,
                pluginName);

        switch (pluginType) {
            case READER:
                return new FastReaderRunner(taskPlugin);
            case WRITER:
                return new FastWriterRunner(taskPlugin);
            default:
                throw DataXException.asDataXException(
                        FrameworkErrorCode.RUNTIME_ERROR,
                        String.format("插件[%s]的类型必须是[reader]或[writer]!",
                                pluginName));
        }
    }

    /**
	 * 反射出具体plugin实例
	 *
	 * @param pluginType
	 * @param pluginName
	 * @param pluginRunType
	 * @return
	 */
	@SuppressWarnings("unchecked")
	private static synchronized Class<? extends AbstractPlugin> loadPluginClass(PluginType pluginType,
			String pluginName, ContainerType pluginRunType) {
		Configuration pluginConf = getPluginConf(pluginType.toString(), pluginName);

//      JarLoader jarLoader = LoadUtil.getJarLoader(pluginType, pluginName);
		try {

//          return (Class<? extends AbstractPlugin>) jarLoader
//                  .loadClass(pluginConf.getString("class") + "$"
//                          + pluginRunType.value());
			return (Class<? extends AbstractPlugin>) Class
					.forName(pluginConf.getString("class") + "$" + pluginRunType.value());

		} catch (Exception e) {
			throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR, e);
		}
	}

	public static DataBaseType getDataBaseType(PluginType pluginType, String pluginName) {
		
		Configuration pluginConf = getPluginConf(pluginType.toString(), pluginName);
		try {
			Field field = Class.forName(pluginConf.getString("class")).getDeclaredField("DATABASE_TYPE");
			field.setAccessible(true);
			DataBaseType dataBaseType = (DataBaseType) field.get(null);
			return dataBaseType;
		} catch (Exception e) {
			throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR, e);
		}
		
		
       
    }
    
}
